package com.frinder.base.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.TimeUnit;

/**
 * @ClassName ConnectionWatchDog
 * @Author frinder
 * @Description netty客户端断线重连机制实现类
 * @Date 2018/11/28 17:32
 * @Version 1.0
 */
@Slf4j
@ChannelHandler.Sharable
public abstract class ConnectionWatchDog extends ChannelInboundHandlerAdapter implements TimerTask {

    private AbstractClient client;
    private final Bootstrap bootstrap;
    private final HashedWheelTimer timer;
    private final int port;
    private final String host;
    private volatile boolean reconnect = true;
    private final int maxAttempts = 166;
    private int attempts;

    public ConnectionWatchDog(AbstractClient client, boolean reconnect) {
        this.client = client;
        this.host = client.getHost();
        this.port = client.getPort();
        this.reconnect = reconnect;
        this.bootstrap = new Bootstrap();
        this.timer = new HashedWheelTimer();
    }

    /**
     * @Author frinder
     * @Description 实现类必须实现，每次重新构建新数组，
     * 数组中元素需要重新 new，否则可能报 @Shareable 异常
     * @Date 2018/11/30 11:04
     * @Param []
     * @Return io.netty.channel.ChannelHandler[]
     */
    public abstract ChannelHandler[] handlers();

    public Bootstrap getBootstrap() {
        return bootstrap;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        attempts = 0;
        ctx.fireChannelActive();
        log.info("*** 客户端 {} 已连接！", client.getName());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.warn("*** 客户端 {} 连接断开！", client.getName());
        if (reconnect) {
            if (attempts < maxAttempts) {
                attempts++;
                int timeout = 2 << attempts;
                log.warn("*** 客户端 {} 连接断开，第 {} 次重连，{} 毫秒后重连超时！",
                        client.getName(), attempts, timeout);
                timer.newTimeout(this, timeout, TimeUnit.MILLISECONDS);
            }
        }
        ctx.fireChannelInactive();
    }

    @Override
    public void run(Timeout timeout) throws Exception {
        ChannelFuture future;
        synchronized (bootstrap) {
            bootstrap.handler(new ChannelInitializer<Channel>() {
                @Override
                protected void initChannel(Channel ch) throws Exception {
                    ch.pipeline().addLast(handlers());
                }
            });
            future = bootstrap.connect(host, port);
        }
        //future对象
        future.addListener(
                (ChannelFuture f) -> {
                    boolean succeed = f.isSuccess();
                    if (!succeed) {
                        log.error("*** 客户端 {} 重连失败！", client.getName());
                        f.channel().pipeline().fireChannelInactive();
                    } else {
                        // reset channel
                        client.resetChannel(f.channel());
                        log.info("*** 客户端 {} 重连成功！", client.getName());
                    }
                }
        );
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
        log.error("*** 客户端 {} 连接异常！！！", client.getName());
    }
}